package com.gtja.beoms.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.github.pagehelper.PageInfo;
import com.gtja.beoms.anno.TimeAnnotation;
import com.gtja.beoms.base.*;
import com.gtja.beoms.config.RedisKeyConfig;
import com.gtja.beoms.dao.*;
import com.gtja.beoms.dto.*;
import com.gtja.beoms.pojo.*;
import com.gtja.beoms.service.IFinanceService;
import com.gtja.beoms.service.IGroupService;
import com.gtja.beoms.thread.MultiThreadFinance;
import com.gtja.beoms.utils.*;
import com.xxl.job.core.context.XxlJobHelper;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.redisson.api.RedissonClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import javax.swing.*;
import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static cn.hutool.core.util.ArrayUtil.append;

//import com.gtja.beoms.thread.MultiThreadFinance;

/**
 * 会话存档
 *
 * @Author : LI
 */
@Service
public class FinanceServiceImpl implements IFinanceService {

    private Logger log = LoggerFactory.getLogger(FinanceServiceImpl.class);

    @Value("${oms.es.databaseName}")
    private String databaseName;

    @Value("${token.key.env}")
    private String keyEnv;

    @Resource
    private QwContactMappingDao qwContactMappingDao;

    @Resource
    private QwWorkMessageDao qwWorkMessageDao;

    @Autowired
    private ElasticSearchUtils elasticSearchUtils;

    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    @Resource
    private QwStaffRelationDao qwStaffRelationDao;

    @Resource
    private QwCustomerDao qwCustomerDao;

    @Resource
    private QwGroupDao qwGroupDao;

    @Resource
    private QwGroupMemberDao qwGroupMemberDao;

    @Resource
    private QwEncryptKeyDao qwEncryptKeyDao;

    @Resource
    private QwAgentInfoDao qwAgentInfoDao;

    @Resource
    private QwAccountDao qwAccountDao;

    @Autowired
    private FinanceUtils financeUtils;

    @Autowired
    private ApiUrlConstants apiUrlConstants;

    @Autowired
    private CosUrlConstants cosUrlConstants;

    @Autowired
    private KafkaUrlConstans kafkaUrlConstans;

    @Autowired
    private IGroupService iGroupChatService;

    @Autowired
    private RedissonClient redissonClient;


    public static ExecutorService executor = ThreadUtil.newExecutor(5);

    /**
     * 一次拉取消息的条数，最大值1000，超过1000返回1000
     */
    public static long TENCENT_SEQ_LIMIT = 500L;

    /**
     * 会话存档同步
     *
     * @param corpId
     * @param methodParams
     * @throws Exception
     */
    @Override
    public void syncFinance(String corpId, String[] methodParams) throws Exception {
        XxlJobHelper.log("执行会话存档定时任务参数: corpId[{}]", corpId);
        Map<String, Object> paramMap = new HashMap<>();
        paramMap.put("corpId", corpId);
        paramMap.put("agentId", AgentConstants.CONVERSATION_AGENTID);
        //查出会话存档secret
        QwAgentInfo qwAgentInfo = qwAgentInfoDao.getAgent(paramMap);
        //创建索引
//        elasticSearchUtils.createIndex2(databaseName, elasticSearchUtils.getFinanceMapping());
        //从缓存中获取消息标识
        String seqKey = keyEnv + "_" + corpId + "_" + "chatSeq";

        String seqStr = redisTemplate.opsForValue().get(seqKey);
        XxlJobHelper.log("本次请求获取消息记录开始的seqStr值: seqStr[{}]", seqStr);
        long seq = StringUtils.isNotEmpty(seqStr) ? Long.valueOf(seqStr) : 0L;
        long startSeq = seq;
        long endSeq = seq;
        XxlJobHelper.log("本次请求获取消息记录开始的seq值: seq[{}]", seq);
        // 调用企微api 整合会话存档数据
        XxlJobHelper.log("corpId是[{}], secret是[{}]", corpId, qwAgentInfo.getAgentSecret());

        //查询私钥及版本号
        QwEncryptKey qwEncryptKey = new QwEncryptKey();
        qwEncryptKey.setCorpId(corpId);
        List<QwEncryptKey> qwEncryptKeys = qwEncryptKeyDao.selectQwEncryptKeyList(qwEncryptKey);
        if (!CollectionUtils.isEmpty(qwEncryptKeys)) {
            String corpName = qwAccountDao.selectOne(new LambdaQueryWrapper<QwAccountInfo>()
                    .eq(QwAccountInfo::getCorpId, corpId))
                    .getCorpName();
            log.info("公司的名称是:[{}]", corpName);
            // 将version和私钥转换成Map
            Map<Long, String> map = qwEncryptKeys.stream().collect(Collectors.toMap(QwEncryptKey::getVersion, QwEncryptKey::getPrivateKey));
            XxlJobHelper.log("会话存档密钥 Map[{}]", map.toString());
            //初始化errorLSet
            Set<String> errorSet = Collections.synchronizedSet(new HashSet<>());
            //循环多次调取腾讯API
            while (true) {
                //如果上次拉去的size小于limit 表示已经结束
                if (checkEndStatus(corpId)) {
                    //先把seqSize归零
                    setSeqSizeZero(corpId);
                    break;
                }
                executor.execute(new MultiThreadFinance(qwAgentInfo.getAgentSecret(), TENCENT_SEQ_LIMIT, map, corpId, methodParams, corpName, keyEnv, redissonClient));
            }
            //等待会话结束,5秒轮询一次
            while (true) {
                ThreadUtil.safeSleep(TimeUnit.SECONDS.toMillis(5));
                log.info("休眠5s等待任务结束");
                if (checkEndStatus(corpId)) {
                    endSeq = getSeq(corpId);
                    log.info("任务结束，退出");
                    //TODO 需要把redisClient重新设计
                    break;
                }
            }

        }
        QwAccountInfo qwAccountInfo = qwAccountDao.selectOne(new LambdaQueryWrapper<QwAccountInfo>()
                .eq(QwAccountInfo::getCorpId, corpId));
        sendSms(methodParams, startSeq, endSeq, qwAccountInfo, corpId);
        XxlJobHelper.log("销毁SDK成功");
    }

    /**
     * 将seqSize归零， 供下次使用
     *
     * @param corpId
     */
    private void setSeqSizeZero(String corpId) {
        String seqSizeKey = keyEnv + corpId + "seqSize";
        redisTemplate.opsForValue().set(seqSizeKey, "0");
    }

    /**
     * 调用cos接口发送短信
     *
     * @param methodParams
     * @param startSeq
     * @param endSeq
     * @param qwAccountInfo
     * @param corpId
     */
    private void sendSms(String[] methodParams, long startSeq, long endSeq, QwAccountInfo qwAccountInfo, String corpId) throws UnsupportedEncodingException {
        //循环除了corpId的手机号
        if (methodParams.length > 2) {
            for (int i = 2; i < methodParams.length; i++) {
                StringBuffer stringBuffer = new StringBuffer();
                //加上公司名字
                if (null != qwAccountInfo) {
                    if (null != qwAccountInfo.getCorpName()) {
                        stringBuffer.append(qwAccountInfo.getCorpName());
                    }
                }
                //成功话术
                RedisUtils redisUtils = new RedisUtils();
                String errorSeqKey = keyEnv + "_" + corpId + "_" + "errorMsgType";
                long errorListSize = redisUtils.sGetSetSize(errorSeqKey);
                if (0 == errorListSize) {
                    stringBuffer.append("-定时任务-会话存档执行成功，任务开始时的SEQ为")
                            .append(startSeq)
                            .append("，任务结束时的SEQ为")
                            .append(endSeq);
                } else {
                    Set<Object> errorSet = redisUtils.sGet(keyEnv + "_" + corpId + "_" + "errorMsgType");
                    //失败话术
                    stringBuffer.append("-定时任务-会话存档以下消息类型请求Kafka异常：")
                            .append(StringUtils.strip(errorSet.toString(), "[]"));
                }
                String url = cosUrlConstants.SEND_SMS;
                String content = stringBuffer.toString();
                HashMap<String, Object> params = new HashMap<>(4);
                params.put("mobile", methodParams[i]);
                params.put("content", content);
                String response = null;
                try {
                    response = HttpRequestUtil.huToolPost(url, params);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                XxlJobHelper.log("请求cos发送短信的url为[{}]", url);
                XxlJobHelper.log("请求cos发送短信的响应结果[{}]", response);
            }
        }
    }


    /**
     * 会话数据插入DB及ES
     *
     * @return
     */
    @TimeAnnotation
    public HashSet<String> FinanceSaveData(List<JSONObject> chatDataList, String corpId, long seq, String flag, HashSet<String> exceptionList, String corpName) throws Exception {
        // 会话数据及会话mapping关系落库
        List<JSONObject> elasticSearchEntities = saveWeChatContactMappingAndWorkMessage(chatDataList, corpId, exceptionList);
        XxlJobHelper.log("本次会话存档插入数据库成功: seq[{}]", seq);
        // 会话数据存入es
        BulkResponse bulkResponse = elasticSearchUtils.insertBatch(databaseName, elasticSearchEntities, corpId);
        // 数据写入Kafka TO 智能质检
        HashSet<String> produceMsgList = new HashSet<>();
        if (CommonConstants.DO_KAFKA_PRODUCER.equals(flag)) {
            XxlJobHelper.log("flag为[{}]执行调用kafka生产者部分", flag);
            produceMsgList = sendWorkMessageToKafka(elasticSearchEntities, corpId, exceptionList, corpName);
        } else {
            XxlJobHelper.log("flag为[{}]不执行调用kafka生产者部分", flag);
        }
        // bulkResponse.hasFailures() false 成功 true 失败
        if (bulkResponse == null || bulkResponse.hasFailures()) {
            XxlJobHelper.log("本次会话存档插入ES失败: seq[{}],e[{}]", seq, bulkResponse.buildFailureMessage());
            //自定义Exception不会回滚数据 ES出错只是让循环停止 会话数据不会回滚
            // bulkResponse.buildFailureMessage()内有seq信息可以查看DB内的当前seq的信息分析错误原因
            throw new RuntimeException();
        }
        XxlJobHelper.log("本次会话存档插入ES成功: seq[{}]", seq);
        return produceMsgList;
    }

    /**
     * 会话数据写入kafka
     */
    @TimeAnnotation
    public HashSet<String> sendWorkMessageToKafka(List<JSONObject> chatDataList, String corpId, HashSet<String> exceptionList, String corpName) {
        //初始化四个不同类型的kafka队列
        QwWorkMessaeWithToListInfoKafkaDTO textTypeList = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_TEXT);
        QwWorkMessaeWithToListInfoKafkaDTO imageTypeList = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_IMAGE);
        QwWorkMessaeWithToListInfoKafkaDTO videoTypeList = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_VIDEO);
        QwWorkMessaeWithToListInfoKafkaDTO voiceTypeList = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_VOICE);
        QwWorkMessaeWithToListInfoKafkaDTO meetingVoiceCallTypeDTO = createNewMsgTypeDTO(CommonConstants.KAFKA_MSG_TYPE_MEETING_VOICE_CALL);
        for (JSONObject chatData : chatDataList) {

            //判断toListType 1是企业员工，2是客户，3是群聊
            Integer seq = chatData.getInteger("seq");
            String msgId = chatData.getString("msgid");
            String action = FinanceConstants.getNameByValue(chatData.getString("action"));
            String from = chatData.getString("from");
            String roomId = chatData.getString("roomid");
            String toList = StringUtils.join(chatData.getJSONArray("tolist"), ",");
            String msgType = chatData.getString("msgtype");
            Date msgTime = new Date(chatData.getLong("msgtime"));
            String fromType = financeUtils.weCustomTypeJudgment(from);
            String toListType;
            if (StringUtils.isEmpty(roomId)) {
                toListType = financeUtils.weCustomTypeJudgment(String.valueOf(toList));
            } else {
                toListType = FinanceConstants.CONTACTS_TYPE_GROUP_CHAT.getType();
            }
            String content = Optional.ofNullable(JSONObject.toJSONString(chatData.getJSONObject(msgType)))
                    .orElse(chatData.getString("content"));

            // 取出formName userCode
            String fromName = null;
            String fromUserCode = null;
            //如果是顾客只查询username
            if (CommonConstants.CUSTOMER_TYPE.equals(fromType)) {
                fromName = qwCustomerDao.selectUserNameById(from, corpId);
            }
            //如果是员工还需要查询usercode
            if (CommonConstants.STAFF_TYPE.equals(fromType)) {
                StaffWorkMessageDTO staffWorkMessageDTO = qwStaffRelationDao.selectUserNameById(from, corpId);
                if (null != staffWorkMessageDTO) {
                    fromName = staffWorkMessageDTO.getUserName();
                    fromUserCode = staffWorkMessageDTO.getUserCode();
                }
            }
            // 取出tolistName userCode
            String tolistName = null;
            String tolistUserCode = null;
            //如果是顾客只查询username
            if (CommonConstants.CUSTOMER_TYPE.equals(toListType)) {
                tolistName = qwCustomerDao.selectUserNameById(toList, corpId);
            }
            //如果是员工还需要查询usercode
            if (CommonConstants.STAFF_TYPE.equals(toListType)) {
                StaffWorkMessageDTO staffWorkMessageDTO = qwStaffRelationDao.selectUserNameById(from, corpId);
                if (null != staffWorkMessageDTO) {
                    tolistName = staffWorkMessageDTO.getUserName();
                    tolistUserCode = staffWorkMessageDTO.getUserCode();
                }
            }

            // 取出群聊信息 ownerName userCode
            String owner = null;
            String ownerName = null;
            String ownerUserCode = null;
            String groupName = null;
            String notice = null;
            String status = null;
            //这个查询是根据员工来的，如果员工没有就需要到客户表中查询
            if (CommonConstants.GROUP_TYPE.equals(toListType)) {
                GroupWIthOwnerNameDTO groupWIthOwnerNameDTO = qwGroupDao.selectQwGroupwithName(roomId);
                if (null != groupWIthOwnerNameDTO) {
                    groupName = groupWIthOwnerNameDTO.getGroupName();
                    notice = groupWIthOwnerNameDTO.getNotice();
                    status = groupWIthOwnerNameDTO.getStatus();
                    if (StrUtil.isBlank(groupWIthOwnerNameDTO.getOwnerName())) {
                        String userName = qwCustomerDao.selectUserNameById(owner, corpId);
                        ownerName = userName;
                    } else {
                        ownerName = groupWIthOwnerNameDTO.getOwnerName();
                        ownerUserCode = groupWIthOwnerNameDTO.getOwnerUserCode();
                    }
                }
            }
            //组装所有的参数
            ArrayList<QwGroupMemberWithTolistInfoDTO> qwGroupMemberWithTolistInfoDTOS = qwGroupMemberDao.selectMembersByChatId(roomId);
            QwGroupWithTolistInfoDTO qwGroupWithTolistInfoDTO = new QwGroupWithTolistInfoDTO();
            qwGroupWithTolistInfoDTO.setOwnerUserCode(ownerUserCode);
            qwGroupWithTolistInfoDTO.setOwner(owner);
            qwGroupWithTolistInfoDTO.setOwnerName(ownerName);
            qwGroupWithTolistInfoDTO.setChatId(roomId);
            qwGroupWithTolistInfoDTO.setCorpId(corpId);
            qwGroupWithTolistInfoDTO.setGroupName(groupName);
            qwGroupWithTolistInfoDTO.setNotice(notice);
            qwGroupWithTolistInfoDTO.setQwGroupMembers(qwGroupMemberWithTolistInfoDTOS);
            qwGroupWithTolistInfoDTO.setStatus(status);
            QwWorkMessageWithTolistInfoDTO qwWorkMessageWithTolistInfoDTO = new QwWorkMessageWithTolistInfoDTO();
            qwWorkMessageWithTolistInfoDTO.setTolistUserCode(tolistUserCode);
            qwWorkMessageWithTolistInfoDTO.setTolistName(tolistName);
            qwWorkMessageWithTolistInfoDTO.setTolist(toList);
            qwWorkMessageWithTolistInfoDTO.setUserCode("");
            qwWorkMessageWithTolistInfoDTO.setFromName(fromName);
            qwWorkMessageWithTolistInfoDTO.setAction(action);
            qwWorkMessageWithTolistInfoDTO.setContent(content);
            qwWorkMessageWithTolistInfoDTO.setCorpId(corpId);
            qwWorkMessageWithTolistInfoDTO.setCorpName(corpName);
            qwWorkMessageWithTolistInfoDTO.setFromId(from);
            qwWorkMessageWithTolistInfoDTO.setFromType(fromType);
            qwWorkMessageWithTolistInfoDTO.setFromUserCode(fromUserCode);
            qwWorkMessageWithTolistInfoDTO.setMsgId(msgId);
            qwWorkMessageWithTolistInfoDTO.setMsgTime(msgTime);
            qwWorkMessageWithTolistInfoDTO.setMsgType(msgType);
            qwWorkMessageWithTolistInfoDTO.setQwGroupDTO(qwGroupWithTolistInfoDTO);
            qwWorkMessageWithTolistInfoDTO.setRoomId(roomId);
            qwWorkMessageWithTolistInfoDTO.setSeq(seq);
            qwWorkMessageWithTolistInfoDTO.setTolistType(toListType);
            // 判断msgType 写入四个list里面
            addQwWorkMessageWithToListInfoDTOToDiffTypeList(qwWorkMessageWithTolistInfoDTO,
                    textTypeList,
                    imageTypeList,
                    videoTypeList,
                    voiceTypeList,
                    meetingVoiceCallTypeDTO);
        }
        produceMsg(textTypeList, exceptionList);
        produceMsg(imageTypeList, exceptionList);
        produceMsg(videoTypeList, exceptionList);
        produceMsg(voiceTypeList, exceptionList);
        produceMsg(meetingVoiceCallTypeDTO, exceptionList);
        return exceptionList;
    }


    private void produceMsg(QwWorkMessaeWithToListInfoKafkaDTO textTypeList, HashSet<String> produceExceptionList) {
        if (null != textTypeList) {
            if (textTypeList.getCONTENT().getRECORD().size() > 0) {
                String jsonString = null;
                try {
                    jsonString = JSON.toJSONString(textTypeList);
                } catch (Exception e) {
                    e.printStackTrace();
                    XxlJobHelper.log("json转换异常[{}]", e.getMessage());
                }
                HashMap<String, Object> map = new HashMap<>(8);
                map.put("key", "qwWorkMessage");
                map.put("value", jsonString);
                XxlJobHelper.log("请求生产者value是：[{}]", jsonString);
                XxlJobHelper.log("请求生产者的url是[{}]", kafkaUrlConstans.PRODUCT_TIME_NODE_URL);
                try {
                    String response = HttpRequestUtil.huToolPost(kafkaUrlConstans.PRODUCT_TIME_NODE_URL, map);
                    if (!StrUtil.contains(response, "true")) {
                        produceExceptionList.add(textTypeList.getMSG_TYPE());
                    }
                    XxlJobHelper.log("请求生产者成功[{}]", response);
                } catch (Exception e) {
                    e.printStackTrace();
                    produceExceptionList.add(textTypeList.getMSG_TYPE());
                    XxlJobHelper.log("请求生产者失败：[{}]", e.getMessage());
                }
            } else {
                XxlJobHelper.log("[{}] 类型消息为空，不发送到kafka", textTypeList.getMSG_TYPE());
            }

        }
    }

    /**
     * 根据不同的msgType，将QwWorkMessageWithToListInfoDTO 放到不同类型的List中
     *
     * @param qwWorkMessageWithTolistInfoDTO 实体类
     * @param textTypeList                   文本类型的list
     * @param imageTypeList                  图像类型的list
     * @param videoTypeList                  视频类型的list
     * @param voiceTypeList                  语音类型的list
     * @param meetingVoiceCallTypeDTO
     */
    private void addQwWorkMessageWithToListInfoDTOToDiffTypeList(QwWorkMessageWithTolistInfoDTO qwWorkMessageWithTolistInfoDTO, QwWorkMessaeWithToListInfoKafkaDTO textTypeList, QwWorkMessaeWithToListInfoKafkaDTO imageTypeList, QwWorkMessaeWithToListInfoKafkaDTO videoTypeList, QwWorkMessaeWithToListInfoKafkaDTO voiceTypeList, QwWorkMessaeWithToListInfoKafkaDTO meetingVoiceCallTypeDTO) {
        String msgType = qwWorkMessageWithTolistInfoDTO.getMsgType();
        switch (msgType) {
            case CommonConstants.QW_MSG_TYPE_TEXT:
                textTypeList.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            case CommonConstants.QW_MSG_TYPE_VIDEO:
                videoTypeList.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            case CommonConstants.QW_MSG_TYPE_IMAGE:
                imageTypeList.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            case CommonConstants.QW_MSG_TYPE_VOICE:
                voiceTypeList.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            case CommonConstants.QW_MSG_TYPE_MEETING_VOICE_CALL:
                meetingVoiceCallTypeDTO.getCONTENT().getRECORD().add(qwWorkMessageWithTolistInfoDTO);
                break;
            default:
        }
    }

    /**
     * 初始化一个QwWorkMessaeWithToListInfoKafkaDTO
     *
     * @return
     */
    private QwWorkMessaeWithToListInfoKafkaDTO createNewMsgTypeDTO(String msgType) {
        QwWorkMessaeWithToListInfoKafkaDTO qwWorkMessaeWithToListInfoKafkaDTO = new QwWorkMessaeWithToListInfoKafkaDTO(msgType);
        qwWorkMessaeWithToListInfoKafkaDTO.setMSG_SUBTYPE("QW_CHAT_RECORD");
        SimpleDateFormat sd = new SimpleDateFormat("yyyyMMdd");
        SimpleDateFormat sd1 = new SimpleDateFormat("HHmmss");
        qwWorkMessaeWithToListInfoKafkaDTO.setSEND_DATE(new Integer(sd.format(new Date())));
        qwWorkMessaeWithToListInfoKafkaDTO.setSEND_TIME(new Integer(sd1.format(new Date())));
        qwWorkMessaeWithToListInfoKafkaDTO.setMSG_ID("qwWorkMessage" + RandomStringUtils.randomAlphanumeric(13));
        List<QwWorkMessageWithTolistInfoDTO> records = new ArrayList<QwWorkMessageWithTolistInfoDTO>();
        ContentDTO contentDTO = new ContentDTO();
        contentDTO.setRECORD(records);
        qwWorkMessaeWithToListInfoKafkaDTO.setCONTENT(contentDTO);
        return qwWorkMessaeWithToListInfoKafkaDTO;
    }

    /**
     * 会话存档ES匹配查询
     *
     * @param query
     * @return
     * @throws Exception
     */
    @Override
    public PageInfo<ConversationArchiveVo> queryElasticSearch(ConversationArchiveQuery query) throws Exception {
        Integer pageNum = query.getPageNum() == null ? 1 : query.getPageNum();
        Integer pageSize = query.getPageSize() == null ? 10 : query.getPageSize();
        SearchSourceBuilder builder = new SearchSourceBuilder();
        int from = (pageNum - 1) * pageSize;
        builder.size(pageSize);
        builder.from(from);
        builder.sort("msgtime", SortOrder.ASC);

        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

        //corpId查询
        if (StringUtils.isNotEmpty(query.getCorpId())) {
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("corpid", query.getCorpId()));
        }
        // msg类型查询
        if (StringUtils.isNotEmpty(query.getMsgType())) {
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("msgtype", query.getMsgType()));
        }
        //发送人Id和接收人Id均不为空表示单聊查询
        if (StringUtils.isNotEmpty(query.getFromId()) && StringUtils.isNotEmpty(query.getReceiveId())) {
            boolQueryBuilder.must(
                    QueryBuilders.boolQuery().should(
                            QueryBuilders.boolQuery().must(QueryBuilders.matchPhraseQuery("from", query.getFromId()))
                                    .must(QueryBuilders.matchPhraseQuery("tolist", query.getReceiveId()))
                    ).should(
                            QueryBuilders.boolQuery().must(QueryBuilders.matchPhraseQuery("from", query.getReceiveId()))
                                    .must(QueryBuilders.matchPhraseQuery("tolist", query.getFromId()))
                    ));
            // 设置为单聊状态
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("roomid", ""));
        } else if (CommonConstants.SINGLE_CHAT_TYPE.equals(query.getTabType())) {
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("roomid", ""));
        } else if (CommonConstants.GROUP_CHAT_TYPE.equals(query.getTabType())) {
            boolQueryBuilder.mustNot(QueryBuilders.matchPhraseQuery("roomid", ""));
        }

        // 发送人Id不为空表示成员查询
        if (StringUtils.isNotEmpty(query.getFromId()) && StringUtils.isEmpty(query.getReceiveId())) {
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("from", query.getFromId()));
        }
        //接收人Id查询 目前不存在接收人查询的情况
        if (StringUtils.isNotEmpty(query.getReceiveId()) && StringUtils.isEmpty(query.getFromId())) {
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("tolist", query.getReceiveId()));
        }
        //群聊Id查询
        if (StringUtils.isNotEmpty(query.getRoomId())) {
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("roomid", query.getRoomId()));
        }
        //关键词查询并高亮显示
        if (StringUtils.isNotEmpty(query.getKeyWord())) {
            boolQueryBuilder.must(QueryBuilders.matchPhraseQuery("text.content", query.getKeyWord()));
            //高亮
            HighlightBuilder highlightBuilder = new HighlightBuilder();
            //设置高亮字段
            highlightBuilder.field("text.content");
            //如果要多个字段高亮,这项要为false
            highlightBuilder.requireFieldMatch(true);
            highlightBuilder.preTags("<span style='color:red'>");
            highlightBuilder.postTags("</span>");
            builder.highlighter(highlightBuilder);
        }
        //时间范围查询
        if (StringUtils.isNotEmpty(query.getBeginDateStr()) && StringUtils.isNotEmpty(query.getEndDateStr())) {
            String beginTime = query.getBeginDateStr() + " 00:00:00";
            String endTime = query.getEndDateStr() + " 00:00:00";
            boolQueryBuilder.filter(QueryBuilders.rangeQuery("msgtime").gte(beginTime).lte(endTime));
        }
        builder.query(boolQueryBuilder);
        PageInfo<ConversationArchiveVo> pageInfo = elasticSearchUtils.searchPage(databaseName, builder, pageNum, pageSize, ConversationArchiveVo.class);
        return pageInfo;
    }

    /**
     * 联系人映射关系及会话存档内容入库
     *
     * @param chatDataList
     * @param corpId
     * @param exceptionList
     */

    @Transactional(rollbackFor = Exception.class)
    @TimeAnnotation
    public List<JSONObject> saveWeChatContactMappingAndWorkMessage(List<JSONObject> chatDataList, String corpId, HashSet<String> exceptionList) throws Exception {
        //es整形数据
        List<JSONObject> elasticSearchEntities = new ArrayList<>();
        for (JSONObject chatData : chatDataList) {
            XxlJobHelper.log("联系人映射关系及会话存档内容入库 chatData[{}]:", chatData.toJSONString());
            Integer seq = chatData.getInteger("seq");
            String msgId = chatData.getString("msgid");
            String action = FinanceConstants.getNameByValue(chatData.getString("action"));
            String from = chatData.getString("from");
            String roomId = chatData.getString("roomid");
            String toList = StringUtils.join(chatData.getJSONArray("tolist"), ",");
            if (StrUtil.isBlank(toList)) {
                XxlJobHelper.log("tolist为空 {}", toList);
                continue;
            }
            XxlJobHelper.log("toList不为空继续执行");
            String msgType = chatData.getString("msgtype");
            Date msgTime = new Date(chatData.getLong("msgtime"));
            String toListType;
            if (StringUtils.isEmpty(roomId) && !toList.contains(",")) {
                toListType = financeUtils.weCustomTypeJudgment(String.valueOf(toList));
            } else {
                toListType = FinanceConstants.CONTACTS_TYPE_GROUP_CHAT.getType();
            }
            String content = Optional.ofNullable(JSONObject.toJSONString(chatData.getJSONObject(msgType)))
                    .orElse(chatData.getString("content"));
            // 会话存档绑定关系入库
            QwContactMapping qwContactMapping = new QwContactMapping();
            qwContactMapping.setCorpId(corpId);
            qwContactMapping.setFromId(from);
            // 发送人类型
            qwContactMapping.setFromType(financeUtils.weCustomTypeJudgment(from));
            //接收人id和类型设定
            qwContactMapping.setReceiveId(toList);
            qwContactMapping.setReceiveType(toListType);
            qwContactMapping.setRoomId(roomId);
            Date date = new Date();
            qwContactMapping.setCreateTime(date);
            qwContactMapping.setLastUpdateTime(date);
            insertWeChatContactMapping(qwContactMapping);

            // 会话存档数据入库
            QwWorkMessage qwWorkMessage = new QwWorkMessage();
            qwWorkMessage.setCorpId(corpId);
            qwWorkMessage.setSeq(seq);
            qwWorkMessage.setMsgId(msgId);
            qwWorkMessage.setAction(action);
            qwWorkMessage.setRoomId(roomId);
            qwWorkMessage.setFromId(from);
            qwWorkMessage.setTolist(toList);
            qwWorkMessage.setTolistType(toListType);
            qwWorkMessage.setMsgType(msgType);
            qwWorkMessage.setContent(content);
            qwWorkMessage.setMsgTime(msgTime);
            qwWorkMessage.setCreateTime(date);
            qwWorkMessage.setLastUpdateTime(date);
            XxlJobHelper.log("插入企业微信工作信息表的数据", qwWorkMessage.toString());
            try {
                insertWeChatWorkMessage(qwWorkMessage);
            } catch (Exception e) {
                e.printStackTrace();
                exceptionList.add(CommonConstants.TO_DB_ERROR);
            }

            // es会话存档数据
            chatData.put("corpid", corpId);
            chatData.put("msgtime", DateFormatUtils.format(msgTime, "yyyy-MM-dd HH:mm:ss"));

            //追加fromName和fromHeadUrl
            String fromName = null;
            String fromHeadUrl = null;
            switch (financeUtils.weCustomTypeJudgment(from)) {
                case CommonConstants.CONTACTS_TYPE_MEMBER:
                    Map<String, String> params = new HashMap<>();
                    params.put("corpId", corpId);
                    params.put("userQwId", from);
                    QwStaffRelation qwStaffRelation = qwStaffRelationDao.getQwStaffRelation(params);
                    if (qwStaffRelation != null) {
                        fromName = qwStaffRelation.getUserName();
                        fromHeadUrl = qwStaffRelation.getUserHeadUrl();
                    } else {
                        fromName = "未绑定的员工";
                    }
                    break;
                case CommonConstants.CONTACTS_TYPE_CUSTOMER:
                    QwCustomer qwCustomer = qwCustomerDao.getDistinctCustomInfo(corpId, from);
                    if (qwCustomer != null) {
                        // 客户表查询客户名称
                        fromName = qwCustomer.getCustName();
                        fromHeadUrl = qwCustomer.getCustHeadUrl();
                    } else if (StringUtils.isNotBlank(roomId)) {
                        // 去群聊查询微信昵称/群昵称
                        QwGroupMember qwGroupMember = qwGroupMemberDao.selectQwGroupMember(roomId, from);
                        XxlJobHelper.log("qwGroupMember为{}", qwGroupMember);
                        if (null != qwGroupMember) {
                            fromName = StringUtils.isNotBlank(qwGroupMember.getNickName()) ? qwGroupMember.getNickName() : qwGroupMember.getUserName();
                        }
                    } else {
                        fromName = "未绑定的客户";
                    }
                    break;
                case CommonConstants.CONTACTS_TYPE_ROBOT:
                    fromName = "机器人";
                    break;
                default:
            }

            chatData.put("fromName", fromName);
            chatData.put("fromHeadUrl", fromHeadUrl);
            elasticSearchEntities.add(chatData);
            XxlJobHelper.log("结束一次入库");
        }
        return elasticSearchEntities;
    }

    /**
     * 新增聊天关系映射
     *
     * @param weChatContactMapping 聊天关系映射
     * @return 结果
     */
    public void insertWeChatContactMapping(QwContactMapping weChatContactMapping) {
        List<QwContactMapping> list = qwContactMappingDao.selectQwContactMappingList(weChatContactMapping);
        if (CollectionUtil.isEmpty(list)) {
            qwContactMappingDao.insertQwContactMapping(weChatContactMapping);
        }
        return;
    }

    /**
     * 新增会话消息
     *
     * @param qwWorkMessage 会话消息
     * @return 结果
     */
    public void insertWeChatWorkMessage(QwWorkMessage qwWorkMessage) {
        QwWorkMessage param = new QwWorkMessage();
        param.setCorpId(qwWorkMessage.getCorpId());
        param.setSeq(qwWorkMessage.getSeq());
        List<QwWorkMessage> list = qwWorkMessageDao.selectQwWorkMessageList(param);
        XxlJobHelper.log("查询得到的会话存档list为{}", list.toString());
        if (CollectionUtil.isEmpty(list)) {
            try {
                qwWorkMessageDao.insertQwWorkMessage(qwWorkMessage);
                XxlJobHelper.log("插入成功");
            } catch (Exception e) {
                XxlJobHelper.log("插入失败，异常信息为：{}", e.getMessage());
            }
        }
        return;
    }


    /**
     * 获取员工聊天映射关系
     *
     * @param corpId
     * @param fromId
     * @return
     * @throws Exception
     */
    @Override
    public List<QwContactMappingVo> getContactMapping(String corpId, String fromId) throws Exception {

        // 查询员工聊天绑定关系
        List<QwContactMappingVo> qwContactMappingVoList = qwContactMappingDao.selectQwContactMappingByUser(corpId, fromId);
        // 循环查询员工头像
        qwContactMappingVoList.stream().forEach(e -> {
            if (CommonConstants.CONTACTS_TYPE_MEMBER.equals(e.getReceiveType())) {
                Map<String, String> params = new HashMap<>();
                params.put("corpId", e.getCorpId());
                params.put("userQwId", e.getReceiveId());
                QwStaffRelation qwStaffRelation = qwStaffRelationDao.getQwStaffRelation(params);
                if (qwStaffRelation != null) {
                    e.setReceiveName(qwStaffRelation.getUserName());
                    e.setReceiveHeadUrl(qwStaffRelation.getUserHeadUrl());
                }
            } else if (CommonConstants.CONTACTS_TYPE_CUSTOMER.equals(e.getReceiveType())) {
                Map<String, String> params = new HashMap<>();
                params.put("corpId", e.getCorpId());
                params.put("custQwId", e.getReceiveId());
                List<QwCustomer> customList = qwCustomerDao.getCustomList(params);
                if (CollectionUtil.isNotEmpty(customList)) {
                    e.setReceiveName(customList.get(0).getCustName());
                    e.setReceiveHeadUrl(customList.get(0).getCustHeadUrl());
                }
            }
        });
        return qwContactMappingVoList;
    }


    /**
     * 获取群聊映射关系
     *
     * @param corpId
     * @param fromId
     * @return
     * @throws Exception
     */
    @Override
    public List<QwContactMappingVo> getGroupMapping(String corpId, String fromId) throws Exception {

        // 查询群聊天绑定关系
        List<QwContactMappingVo> qwContactMappingVoList = qwContactMappingDao.selectQwContactMappingByGroup(corpId, fromId);
        //查询群名字和成员头像
        qwContactMappingVoList.stream().forEach(e -> {
            Map<String, Object> param = new HashMap<>();
            param.put("chatId", e.getRoomId());
            param.put("corpId", corpId);
            // 群聊数据通用
            QwGroupDTO qwGroupDTO;
            try {
                qwGroupDTO = iGroupChatService.getGroupInfo(param);
            } catch (Exception exception) {
                throw new RuntimeException(exception);
            }
            if (qwGroupDTO != null) {
                // 成员头像取出 用于拼接群聊头像
                if (CollectionUtil.isNotEmpty(qwGroupDTO.getQwGroupMembers())) {
                    // 排序 HeadUrl为null置后
                    qwGroupDTO.getQwGroupMembers().sort(Comparator.comparing(QwGroupMemberDTO::getUserHeadUrl, Comparator.nullsLast(String::compareTo)));
                    e.setQwGroupMembers(qwGroupDTO.getQwGroupMembers());
                }
            }
        });
        return qwContactMappingVoList;
    }


    /**
     * 获取有会话记录的员工信息
     *
     * @param corpId
     * @return
     * @throws Exception
     */
    @Override
    public List<QwContactMappingVo> getUserInfo(String corpId) throws Exception {
        // 查询企业有聊天记录的员工信息
        List<QwContactMappingVo> qwContactMappingVos = qwContactMappingDao.selectQwContactMappingUserInfo(corpId, CommonConstants.CONTACTS_TYPE_MEMBER);
        // 查询员工的姓名和头像和群聊数和个聊数
        qwContactMappingVos.forEach(e -> {
            //查询员工的姓名和头像
            Map<String, String> params = new HashMap<>();
            params.put("corpId", e.getCorpId());
            params.put("userQwId", e.getFromId());
            QwStaffRelation qwStaffRelation = qwStaffRelationDao.getQwStaffRelation(params);
            if (qwStaffRelation != null) {
                e.setFromName(qwStaffRelation.getUserName());
                e.setFromHeadUrl(qwStaffRelation.getUserHeadUrl());
            }
            // 查询员工的个聊数
            List<QwContactMappingVo> singleChatList = qwContactMappingDao.selectQwContactMappingByUser(e.getCorpId(), e.getFromId());
            e.setTotalSingleChat(CollectionUtils.isEmpty(singleChatList) ? 0 : singleChatList.size());
            // 查询员工的群聊数
            List<QwContactMappingVo> groupChatList = qwContactMappingDao.selectQwContactMappingByGroup(e.getCorpId(), e.getFromId());
            e.setTotalGroupChat(CollectionUtils.isEmpty(groupChatList) ? 0 : groupChatList.size());
        });
        return qwContactMappingVos;
    }

    /**
     * 拿到seq
     *
     * @return
     */
    private int getSeq(String corpId) {
        String seqKey = keyEnv + "_" + corpId + "_" + "chatSeq";
        String seqStr = redisTemplate.opsForValue().get(seqKey);
        return Integer.parseInt(seqStr);
    }

    /**
     * 如果seqSize小于limit表示已经取到了最后，并且seq不是0，可以退出了
     *
     * @return
     */
    private boolean checkEndStatus(String corpId) {
        String seqSizeKey = keyEnv + corpId + "seqSize";
        String seqSizeStr = redisTemplate.opsForValue().get(seqSizeKey);
        int seqSize = StringUtils.isNotEmpty(seqSizeStr) ? Integer.parseInt(seqSizeStr) : 0;

        String seqKey = keyEnv + "_" + corpId + "_" + "chatSeq";
        String seqStr = redisTemplate.opsForValue().get(seqKey);
        int seq = StringUtils.isNotEmpty(seqStr) ? Integer.parseInt(seqStr) : 0;

        return seqSize < TENCENT_SEQ_LIMIT && seq > 0 && seqSize > 0;
    }

}
